package cn.lecosa.spark.broadcast

import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.broadcast.Broadcast

object MapSideJoin {
/* 模拟MR的map sise join 有一个小文件 这个小文件和大文件进行join 默认情况下会进行shuffle
 * 可以将小文件缓存到map端 其实就是将小文件载入到内存
*/  
  def main(args: Array[String]): Unit = {
    val conf =new SparkConf().setMaster("local[*]").setAppName("braodcastapp");
     val sc:SparkContext =new SparkContext(conf);
     val rules =Map("cn"->"中国","us"->"美国","jp"->"日本")

    //将driver端的变量 广播到属于自己 的所有executor
    val lines = sc.textFile("hdfs://park01:9000/home/broadcast.txt", 2)
    val broadcast: Broadcast[Map[String, String]] = sc.broadcast(rules)
    val linesbc= sc.broadcast(lines);
    val  res= lines.map { line => {
       val fields =line.split(",")
       val name =fields(0)
       val nation_code=fields(1)
//       val nation_name=rules(nation_code)
//在executor端 拿到广播变量中的值       
      val r: Map[String, String]=broadcast.value;
       val nation_name=r(nation_code)
//      val brval= linesbc.value
       (name,nation_name)
     } }
     
     print(res.collect().toBuffer)
     /*函数在driver端定义 在exector的task中执行
      * 
      * 如果在driver端创建的一个变量 ，并且在rdd 方法的函数中
      * rdd方法中传入的函数是在executor的他上课中执行
      * dirver会将这个在driver 中定义的变量发送给每一个task
      * 后面生成的task也会从新生成
      * 
*/     
     
     
  }
}